package com.atguigu.edu.app.dim;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONException;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.app.func.MyBroadcastFunction;
import com.atguigu.edu.app.func.MyPhoenixSink;
import com.atguigu.edu.bean.TableProcess;
import com.atguigu.edu.util.KafkaUtil;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.streaming.api.datastream.BroadcastConnectedStream;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

public class DimApp {
    public static void main(String[] args) throws Exception {
        // TODO 1 环境准备
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        // TODO 2 设置状态后端
                /*
                env.enableCheckpointing(5 * 60 * 1000L, CheckpointingMode.EXACTLY_ONCE );
                env.getCheckpointConfig().setCheckpointTimeout( 3 * 60 * 1000L );
                env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);
                env.setStateBackend(new HashMapStateBackend());
                env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/gmall/ck");
                System.setProperty("HADOOP_USER_NAME", "atguigu");
                */

        // TODO 3 读取kafka的数据
        String topicName = "topic_db";
        String groupId = "dim_app_0409";
        DataStreamSource<String> dbStream = env.addSource(KafkaUtil.getFlinkKafkaConsumer(topicName, groupId));

      // dbStream.print("db>>>");

        // TODO 4 过滤脏数据并转换为jsonObject
        // 过滤"type"="bootstrap-start"||"bootstrap-complete"
        OutputTag<String> dirtyOutputTag = new OutputTag<String>("dirty") {
        };
        SingleOutputStreamOperator<JSONObject> processStream = dbStream.process(new ProcessFunction<String, JSONObject>() {
            @Override
            //TODO   XXXXXXXXXXXXXXXXXXXXX没写完
            public void processElement(String s, Context context, Collector<JSONObject> collector) throws Exception {
                try {
                    JSONObject jsonObject = JSON.parseObject(s);
                    String type = jsonObject.getString("type");
                    if (!("bootstrap-start".equals(type) || "bootstrap-complete".equals(type))) {
                        collector.collect(jsonObject);
                    } else {
                        //特殊类型没有意义的数据写入测输出流
                        context.output(dirtyOutputTag, s);
                    }
                } catch (JSONException e) {
                    e.printStackTrace();
                    //不为JSON写到测输出流
                    context.output(dirtyOutputTag, s);
                }
            }
        });

//        processStream.print("proc>>>>");

        //TODO 5 使用flinkCDC实时监控配置表
        MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
                .username("root")
                .password("123456")
                .hostname("hadoop102")
                .port(3306)
                .databaseList("realtime_config")
                .tableList("realtime_config.table_process")
                .startupOptions(StartupOptions.initial())
                .deserializer(new JsonDebeziumDeserializationSchema())
                .build();

        DataStreamSource<String> flinkCDC = env.fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "flinkCDC");

        MapStateDescriptor<String, TableProcess> mapStateDescriptor = new MapStateDescriptor<>("table_process", String.class, TableProcess.class);
        BroadcastStream<String> broadcastStream = flinkCDC.broadcast(mapStateDescriptor);

        // TODO 6 连接主流和广播流
        BroadcastConnectedStream<JSONObject, String> connectedStream = processStream.connect(broadcastStream);

        // TODO 7 处理连接流
        SingleOutputStreamOperator<JSONObject> tableProcessStream = connectedStream.process(new MyBroadcastFunction(mapStateDescriptor));

        //tableProcessStream.print("connect>>>");
        // TODO 8 数据写出到hbase
        tableProcessStream.addSink(new MyPhoenixSink());
        // TODO 9 执行
        env.execute(groupId);
    }
}
